package com.shujia.sql

import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.StateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.catalog.hive.HiveCatalog

object Demo07OnHive {
  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    bsEnv.setParallelism(1)

    // 每1000ms做一次checkpoint
    bsEnv.enableCheckpointing(10000)
    // 高级选项(可选)
    // 设置CheckPoint的模式为EXACTLY_ONCE精确一次/完全一次(默认)
    bsEnv.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 设置两个CheckPoint任务之间的时间间隔
    bsEnv.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // 设置CheckPoint的超时时间
    bsEnv.getCheckpointConfig.setCheckpointTimeout(60000)
    // 设置同一时刻最多能有多少个CheckPoint任务
    bsEnv.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    // 设置在任务取消时不清理CheckPoint保存的状态
    bsEnv.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    // 设置CheckPoint目录 使用文件系统作为状态后端（保存状态的地方）
    val backend: StateBackend = new RocksDBStateBackend("hdfs://master:9000/flink/wc/checkpoint", true)
    bsEnv.setStateBackend(backend)

    val name = "myhive"
    val defaultDatabase = "flink"
    val hiveConfDir = "/usr/local/soft/hive-1.2.1/conf" // hive配置文件的路径

    val hive = new HiveCatalog(name, defaultDatabase, hiveConfDir)
    bsTableEnv.registerCatalog("myhive", hive)

    // set the HiveCatalog as the current catalog of the session
    bsTableEnv.useCatalog("myhive")

    // 切换元数据
    bsTableEnv.useCatalog("myhive")

    bsTableEnv.executeSql(
      """
        |insert into city_cnt
        |select city_id,count(distinct mdn) from dianxin group by city_id
        |""".stripMargin)
  }

}
